package com.siteseat.crudeq.kernel;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.siteseat.common.foundation.PropertyHelper;
import com.siteseat.crudeq.common.raw.Constants;
import com.siteseat.crudeq.kernel.exception.CreateChunkFailureException;
import com.siteseat.crudeq.kernel.exception.InvalidChunkException;

/**
 *
 */
public class ChunkManager extends KernelComponent implements Constants {
  
  private QueueProfile           profile;

  private AtomicReference<Chunk> active       = new AtomicReference<Chunk>();

  private ArrayList<ChunkInfo>   all;

  private final ReadWriteLock    activeLock   = new ReentrantReadWriteLock();

  private Map<Long, Chunk>       offlined     = new HashMap<Long, Chunk>();

  private final ReadWriteLock    offlinedLock = new ReentrantReadWriteLock();

  /**
   * @param profile
   */
  public ChunkManager( QueueProfile profile ) {
    this.profile = profile;
  }

  protected ArrayList<ChunkInfo>  getAllChunkInfo() {
    return all;
  }
  
  /**
   * Gets the active
   *
   * @return the active
   */
  protected Chunk getActiveChunk() {
    return active.get();
  }
  
  protected void setActiveChunk( Chunk c ) {
    active.set( c );
  }

  /**
   * Gets the activeLock
   *
   * @return the activeLock
   */
  protected ReadWriteLock getActiveChunkLock() {
    return activeLock;
  }

  /**
   * Gets the offlined
   *
   * @return the offlined
   */
  protected Map<Long, Chunk> getOfflinedChunks() {
    return offlined;
  }

  /**
   * Gets the offlinedLock
   *
   * @return the offlinedLock
   */
  protected ReadWriteLock getOfflinedChunkLock() {
    return offlinedLock;
  }

  /**
   * Sets the all
   *
   * @param all the all to set
   */
  public void setAllChunkInfo( ArrayList<ChunkInfo> all ) {
    this.all = all;
  }

  /**
   * @return
   */
  public long getFirstChunkId() {
    ChunkInfo ci = this.all.get( 0 );
    return ci.getChunkId().longValue();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doInit() throws Exception {
    super.doInit();
    
    PropertyHelper.ensureInitialized( log, getProfile(), "profile" );

    load();
  }

  /**
   * Gets the profile
   * 
   * @return the profile
   */
  public QueueProfile getProfile() {
    return profile;
  }

  /**
   * @return
   */
  public Integer getQueueId() {
    return getProfile().getId();
  }

  /**
   * @return
   * @throws IOException
   * @throws InvalidChunkException
   * @throws IOException
   * @throws CreateChunkFailureException
   * @throws InterruptedException
   */
  protected void load() throws IOException, CreateChunkFailureException {
    QueueProfile prof = getProfile();

    this.all = new ArrayList<ChunkInfo>();
    
    Long currCid = Long.valueOf( 1 );
    Chunk c = newChunkObject( currCid, profile.getChunkSizeByBytes() );

    this.active.set( c );
    this.all.add( c.getInfo() );

    prof.setCheckChunk( true );
  }

  /**
   * 
   */
  public void checkAllChunks() throws IOException {
    for( ChunkInfo ci: this.all ) {
      checkChunk( ci );
    }
  }

  /**
   * @param ci
   * @throws IOException
   */
  public void checkChunk( ChunkInfo ci ) throws IOException {
    RandomAccessFile raf = new RandomAccessFile( ci.getFile(), "r" );
    byte[] buf = new byte[ci.getSize()];
    try {
      raf.readFully( buf );
    } finally {
      raf.close();
    }

  }

  /**
   * @param messageSize
   * @return
   * @throws InterruptedException
   */
  public Message prepareMessage( int messageSize ) throws InterruptedException, IOException {

    QueueProfile p = getProfile();

    // TODO: restrict the message size

    // This loop alleviates the need for synchronization on the whole ChunkPool
    do {
      Chunk current = this.active.get();
      {
        Message r = current.prepareMessage( messageSize );
        if( r != null ) {
          return r;
        }
      }

      Long currentChunkId = current.getInfo().getChunkId();

      Lock alk = this.activeLock.writeLock();
      alk.lockInterruptibly();
      try {
        Long newChunkId = Long.valueOf( currentChunkId.longValue() + 1 );
        int newCapacity = Math.max( messageSize, p.getChunkSizeByBytes() );
        Chunk newChunk = newChunkObject( newChunkId, newCapacity );
        if( this.active.compareAndSet( current, newChunk ) ) {
          this.all.add( newChunk.getInfo() );
        }
      } finally {
        alk.unlock();
      }

      Lock slk = this.offlinedLock.writeLock();
      slk.lockInterruptibly();
      try {
        if( current.notAllSaved() ) {
          this.offlined.put( currentChunkId, current );
        }
      } finally {
        slk.unlock();
      }

    } while( true );
  }

  /**
   * @param cid
   * @return
   * @throws IOException
   * @throws IOException
   */
  protected Chunk newChunkObject( Long cid, int capacity ) throws IOException {
    KernelFactory f = getFactory();
    ChunkInfo ci = f.newChunkInfo( getProfile(), cid );
    return f.newChunk( ci, capacity );
  }



  /**
   * @param cid
   * @return
   * @throws InvalidChunkException
   * @throws IOException
   * @throws InterruptedException
   */
  protected int historyPosition( MessageId msgId ) throws InvalidChunkException {
    long expectedChunkId = msgId.getChunkId();
    
    int allSize = this.all.size();
    int position;
    int lastPosition = allSize - 1;
    ChunkInfo last = this.all.get( lastPosition );
    ChunkInfo beginFrom;
    
    //Lock alk = this.activeLock.writeLock();
    //alk.lockInterruptibly();
    //try {
    if( expectedChunkId == last.getChunkId().longValue() ) {
      position = allSize - 1;
      beginFrom = last;
    } 
    else {
      int low = 0;
      int high = lastPosition;
      position = -1;
  
      if( expectedChunkId == 0 ) {
        position = high;// start fetching from latest chunk
      } else if( expectedChunkId < 0 ) {
        position = low;
      } else {
        while( low <= high ) {
          int mid = ( low + high ) >>> 1;
          long midChunkId = this.all.get( mid ).getChunkId().longValue();
    
          if( expectedChunkId > midChunkId ) {
            low = mid + 1;
          } else if( expectedChunkId < midChunkId ) {
            high = mid - 1;
          } else {
            position = mid; // key found
            break;
          }
        }
    
        if( position < 0 ) {
          throw new InvalidChunkException();
        }
      }
  
      beginFrom = this.all.get( position );
    }
    
    int offset = msgId.getOffset();
    if( offset < 0 ) {
      throw new InvalidChunkException();
    }

    int beginFromSize = beginFrom.getSize();
    boolean outOfOffset = ( beginFromSize <= offset || beginFromSize < offset + MESSAGE.MINIMAL_SIZE );
    if( outOfOffset ) {
      if( beginFrom == last ) {
        return -1;// at the tail
      }
      if( position >= this.all.size() ) {
        throw new InvalidChunkException();
      }
      // automatically switch to next chunk
      position++;
      offset = 0;
    }
    
    msgId.setOffset( offset );

    return position;
  }
  
  /**
   * @param cid
   * @return
   * @throws InvalidChunkException
   * @throws IOException
   * @throws InterruptedException
   */
  public MessageHistory history( MessageId msgId ) throws InvalidChunkException {
    int position = historyPosition( msgId );
    if( position < 0 ) {
      return null;
    }
    
    Integer qid = msgId.getQueueId();
    int offset = msgId.getOffset();
    
    return getFactory().newMessageHistory( qid, this.all.subList( position, this.all.size() ), offset );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doDestroy() throws Exception {
    // nothing to do
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void setBeanName( String beanName ) {
    super.setBeanName( beanName + "*" + getProfile().getId() );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean equals( Object obj ) {
    ChunkManager that;
    try {
      that = (ChunkManager)obj;
    } catch( ClassCastException e ) {
      return false;
    }
    return getQueueId().equals( that.getQueueId() );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public int hashCode() {
    return getQueueId().hashCode();
  }

}
